package com.hu.flink12.api.source;

import com.hu.flink12.api.entity.Order;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

/**
 * @Author: hujianjun
 * @Date: 2021/2/3 22:45
 * @Describe: 自定义每5s查询数据
 */
public class CustomerMysql {
    public static void main(String[] args) throws Exception {
        //TODO 1.获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        //TODO 2.读取数据源
        DataStream<Order> mysqlDataStream = env.addSource(new CustomerMysqlSource()).setParallelism(1);

        //TODO 3.转换操作

        //TODO 4.sink操作
        mysqlDataStream.print();

        //TODO 5.执行程序
        env.execute();
    }

    public static class CustomerMysqlSource extends RichParallelSourceFunction<Order> {

        private Boolean flag = true;
        private Connection conn = null;
        private PreparedStatement ps = null;
        private ResultSet resultSet = null;

        @Override
        public void open(Configuration parameters) throws Exception {
//            Class.forName("com.mysql.cj.jdbc.Driver");
            conn = DriverManager.getConnection("jdbc:mysql://localhost:3306/test?useUnicode=true&serverTimezone=UTC&useSSL=false&allowPublicKeyRetrieval=true", "root", "root");
            ps = conn.prepareStatement("select * from t_order");
        }

        @Override
        public void close() throws Exception {
            if (conn != null) conn.close();
            if (ps != null) ps.close();
        }

        @Override
        public void run(SourceContext<Order> ctx) throws Exception {
            while (flag) {
                Thread.sleep(10000);
                resultSet = ps.executeQuery();
                while (resultSet.next()) {
                    ctx.collect(new Order(resultSet.getString(1), resultSet.getInt(2), resultSet.getInt(3), resultSet.getLong(4)));
                }
            }
        }

        @Override
        public void cancel() {
            flag = false;
        }
    }
}
